#pragma once

#include <gflags/gflags.h>
#include <thread>
#include <cctype>
#include<brpc/server.h>
#include<butil/logging.h>

#include "utils.hpp"// 基础工具接口
#include "mysql_user.hpp"// mysql数据管理客户端封装
#include "channel.hpp"  // 信道管理模块封装
#include "etcd.hpp"    // 服务注册模块封装  
#include "RabbitMQ.hpp"
#include "mysql_session_member.hpp" 

#include "base.pb.h"// protobuf框架代码
#include "user.pb.h"
#include "transmit.pb.h"


namespace zzy_mims
{
     class MsgTransmitServiceImpl:public MsgTransmitService
    {

    public:
        MsgTransmitServiceImpl(const std::string &user_service_name,
            const ServChanManager::ptr &channels,
            const std::shared_ptr<odb::core::database> &mysql_client,
            const std::string &exchange_name,
            const std::string &routing_key,
            const RMQClient::ptr &mq_client):
            _user_service_name(user_service_name),
            _mm_channels(channels),
            _mysql_session_member_table(std::make_shared<SessionMemberTable>(mysql_client)),
            _exchange_name(exchange_name),
            _routing_key(routing_key),
            _mq_client(mq_client){};
        ~MsgTransmitServiceImpl(){};
        virtual void GetTransmitTarget(google::protobuf::RpcController* controller,
            const ::zzy_mims::NewMessageReq* request, 
            ::zzy_mims::GetTransmitTargetRsp* response, 
            ::google::protobuf::Closure* done)
        {    
            //取出数据，检查数据，插入数据，填写回应
            brpc::ClosureGuard rpc_guard(done);//和RAII思想类似，在局部域结束后自动调用done->Run();
            response->set_request_id(request->request_id());
            auto err_handler = [this,response](const std::string& rid,const std::string& errmsg)
            {
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };
            //1.进行消息组织
            std::string uid = request->user_id();
            std::string csid = request->chat_session_id();

            const MessageContent& msg_cont = request->message();

                //a.调用用户微服务获取用户信息
                auto channel = _mm_channels->choose(_user_service_name);
                if(!channel)
                {
                    LOG_ERROR("没有可供访问的用户微服务节点！{}-{}\n",request->request_id(),_user_service_name);
                    err_handler(request->request_id(),"没有可供访问的用户微服务节点！");
                    return;
                }
                //1.获取rpc服务调用句柄
                ::zzy_mims::UserService_Stub stub(channel.get());
                //2.填写请求
                zzy_mims::GetUserInfoReq req;
                zzy_mims::GetUserInfoRsp resp;
                
                req.set_request_id(request->request_id());
                req.set_user_id(uid);

                //3.调用服务
                brpc::Controller cntl;
                stub.GetUserInfo(&cntl,&req,&resp,nullptr);//同步调用

                //4.验证服务调用是否成功
                if (cntl.Failed() == true || resp.success() == false) 
                {
                    LOG_ERROR("用户子服务调用失败{}-{}!\n", request->request_id(), cntl.ErrorText());
                    err_handler(request->request_id(), "用户子服务调用失败!");
                    return;
                };
            MessageInfo msg_info;
            msg_info.set_message_id(uuid());
            msg_info.set_chat_session_id(csid);
            msg_info.set_timestamp(time(nullptr));
            msg_info.mutable_sender()->CopyFrom(resp.user_info());
            msg_info.mutable_message()->CopyFrom(msg_cont);

            //2.获取需要进行转发的用户id列表
            std::vector<std::string> user_list = _mysql_session_member_table->members(csid);

            //3.将消息发布到rabbitmq中，进行消息持久化
            bool ret = _mq_client->publish(_exchange_name,msg_info.SerializeAsString(),_routing_key);
            if(ret == false)
            {
                LOG_ERROR("向rabbitmq发布消息失败{}-{}\n",request->request_id(),cntl.ErrorText());
                err_handler(request->request_id(),"向rabbitmq发布消息失败");
                return;
            }

            //4.组织回应
            response->set_success(true);
            response->mutable_message()->CopyFrom(msg_info);
            for(auto &uid:user_list)
            response->add_target_id_list(uid);
        
        }
    private:
        //mysql库中聊天会话成员表管理句柄
        SessionMemberTable::ptr _mysql_session_member_table;

        //服务操作句柄
        ServChanManager::ptr _mm_channels;
        std::string _user_service_name;
        
        //rabbitmq句柄
        RMQClient::ptr _mq_client;
        std::string _exchange_name;
        std::string _routing_key;
    };
    class MsgTransmitServer
    {
    public:
        using ptr = std::shared_ptr<MsgTransmitServer>;
        MsgTransmitServer(const Discovery::ptr& service_discoverer,const std::shared_ptr<brpc::Server>& rpc_server)
        :_service_discoverer(service_discoverer),_rpc_server(rpc_server){};
        ~MsgTransmitServer(){};
        void start()
        {
            _rpc_server->RunUntilAskedToQuit();
        }
    private:
        Discovery::ptr _service_discoverer;
        std::shared_ptr<brpc::Server> _rpc_server;
    };
    class MsgTransmitServerBuilder
    {
    public:
        void make_mysql_object
            (const std::string& user,
            const std::string& password,
            const std::string& db,
            const std::string& host,
            const std::string& charset,
            const int port,const int conn_max_pool)
        {
            _mysql_client = ODBFactory::create(user,password,db,host,charset,port,conn_max_pool);
        }
        void make_discover_object(const std::string& etcd_host,
        const std::string& base_service_name,const std::string& user_service_name)
        {
            _user_service_name = user_service_name;
            _mm_channels = std::make_shared<ServChanManager>();
            _mm_channels->Declared(_user_service_name);
            auto online = bind(&zzy_mims::ServChanManager::ServiceOnline,_mm_channels.get(),std::placeholders::_1,std::placeholders::_2);
            auto offline = bind(&zzy_mims::ServChanManager::ServiceOffline,_mm_channels.get(),std::placeholders::_1,std::placeholders::_2);
            _service_discoverer = std::make_shared<zzy_mims::Discovery>(etcd_host,base_service_name,online,offline);
        }

        void make_reg_object(const std::string& reg_host,
            const std::string& service_name,
            const std::string& access_host)
        {
            _reg_client = std::make_shared<Registery>(reg_host);
            _reg_client->Register(service_name,access_host);
        }
        void make_mq_object(const std::string& user,
                            const std::string& password,
                            const std::string& host,
                            const std::string& exchange_name, 
                            const std::string& queue_name,
                            const std::string& binding_key)
        {
            _routing_key = binding_key;
            _exchange_name = exchange_name;
            _mq_client = std::make_shared<RMQClient>(user,password,host);
            _mq_client->declareComponents(exchange_name,queue_name,binding_key);
        }
        void make_rpc_server(uint16_t port, int32_t timeout, uint8_t num_threads)
        {
            if(!_mq_client)
            {
                LOG_ERROR("未初始化rabbitmq客户端\n");
                abort();
            }
            if(!_mysql_client)
            {
                LOG_ERROR("构造rpc服务时,还未初始化MYSQL客户端\n");
                abort();
            }
            if(!_mm_channels)
            {
                LOG_ERROR("未初始化信道管理模块\n");
                abort();
            }
            _rpc_server = std::make_shared<brpc::Server>();
            MsgTransmitServiceImpl* transmit_service = new MsgTransmitServiceImpl(_user_service_name, _mm_channels,
                _mysql_client, _exchange_name, _routing_key, _mq_client);
            int ret = _rpc_server->AddService(transmit_service,brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
            if(ret==-1)
            {
                LOG_ERROR("添加RPC服务失败!");
                abort();
            }
            brpc::ServerOptions options;
            options.idle_timeout_sec=timeout;
            options.num_threads = num_threads;
            ret = _rpc_server->Start(port,&options);
            if(ret==-1)
            {
                LOG_ERROR("transmit_RPC服务器启动失败");
                abort();
            }
        }
        MsgTransmitServer::ptr build()
        {
            if(!_rpc_server)
            {
                LOG_ERROR("构造TransmitServer时,还未初始化rpc客户端\n");
                abort();
            }
            if(!_service_discoverer)
            {
                LOG_ERROR("构造TransmitServer时,还未初始化discover客户端\n");
                abort();
            }
            if(!_reg_client)
            {
                LOG_ERROR("构造TransmitServer时,还未初始化registry客户端\n");
                abort();
            }
            MsgTransmitServer::ptr server = std::make_shared<MsgTransmitServer>(_service_discoverer,_rpc_server);
            return server;
        }

    private:
    //服务发现
        Registery::ptr _reg_client;
        std::string _user_service_name;
        ServChanManager::ptr _mm_channels;
        std::shared_ptr<brpc::Server> _rpc_server;
        Discovery::ptr _service_discoverer;

    //rabbitmq相关
        std::string _routing_key;
        std::string _exchange_name;
        RMQClient::ptr _mq_client;
    
    
        std::shared_ptr<odb::core::database> _mysql_client;

    };

}